<?php
/**
 * class class
 *
 * Description for class class
 *
 * @author: sTalkEr
 *
 **/

if (!defined("MD_READ"))
   define("MD_READ", 1);

if (!defined("MD_SEND"))
   define("MD_SEND", 2);

if (!defined("MD_EXCPECT"))
   define("MD_EXCPECT", 4);

if (!defined("MD_DEBUG"))
   define("MD_DEBUG", true);

if (!defined("MD_REQUEST"))
   define("MD_REQUEST", "GET %s HTTP/1.1\r\nAccept: */*\r\nAccept-Language: en-us\r\nPragma: no-cache\r\nUser-Agent: Monzilla/4.0 (compatible; MSIE 6.0)\r\nHost: %s\r\nConnection: Close\r\n\r\n");



class Downloader
{
   protected $supportProtocols = array("http" => "DownloadViaHTTP");
   protected $timeout;
   protected $maxThread;
   protected $logs = array();
   protected $urls = array();
   protected $contents = array();

   private $allStreams = array();
   private $toRead = array();
   private $toWrite = array();
   private $toException = array();

   private $stats = array();

   public function __construct($timeout = 30, $maxThread = 50, $log = null)
   {
      if ($timeout <= 0)
      {
         throw new Exception("The parametre 'timeout' should be more than 0");
      }
      $this->timeout = $timeout;

      if ($maxThread <= 0)
      {
         throw new Exception("The parametre 'maxThread' should be more than 0");
      }
      $this->maxThread = $maxThread;
      if ($log !== null)
      {
         $this->AddLogEngine($log);
      }
   }

   public function AddLogging($log)
   {
      if (is_array($log))
      {
         foreach ($log as $val)
         {
            $this->AddLogging($val);
         }
      }
      if (is_resource($log))
      {
         $meta = stream_get_meta_data($log);
         if (array_search($meta["mode"], array("w", "wb", "a", "ab")))
         {
            $this->logs[] = $log;
         }
         else
         {
            throw new Exception("The stream should be in writing mode");
         }
      }
      elseif (is_string($log))
      {
         if (($fp = fopen($log, "a")) !== false)
         {
            $this->logs[] = $fp;
         }
         else
         {
            throw new Exception("It was not possible to open file '$log'");
         }
      }
   }

   public function ToLog($message, $n = true)
   {
      foreach ($this->logs as $log)
      {
         fwrite($log, date("r").": $message".($n ? "\n" : ""));
      }
   }

   private function CreateStream($url)
   {
      $info = parse_url($url);
      if (empty($info["host"]))
      {
         $this->ToLog("Warning: URL '$url' not the correct");
         return;
      }
      if (!isset($info["scheme"]))
      {
         $info["scheme"] = "http";
      }
      else
      {
         $info["scheme"] = strtolower($info["scheme"]);
      }
      if (!isset($this->supportProtocols[$info["scheme"]]))
      {
         $this->ToLog("Warning: {$info["scheme"]} protocol isn't supported");
         return;
      }
      if (isset($this->allStreams[$url]))
      {
         $this->ToLog("Warning: URL $url already is present");
         return;
      }

      $port = isset($info["port"]) ? $info["port"] : 80;

      $stream = stream_socket_client(
            "tcp://{$info["host"]}:$port",
            $errno, $errstr, $this->timeout,
            STREAM_CLIENT_ASYNC_CONNECT|STREAM_CLIENT_CONNECT);

      if ($errno == 0)
      {
         stream_set_blocking($stream, 0);
         $num = count($this->allStreams);
         $this->allStreams[$num] = $this->toWrite[] = $this->toException[] = $stream;
         $this->urls[$num] = $info;
         $this->urls[$num]["num"] = $num;
         $this->urls[$num]["code"] = 0;
         $this->urls[$num]["stream"] = $stream;
         $this->urls[$num]["full"] = $url;
         $this->urls[$num]["contents"] = "";
         $this->urls[$num]["headers"] = array();
         return $num;
      }
      else
      {
         $this->ToLog("Error: There was error at stream creation. Error message: $errstr($errno).");
      }
   }

   public function DownloadToArray(array $urls)
   {
      if (count($urls) == 0)
      {
         $this->ToLog("Warning: URL list is empty");
         return array();
      }

      $this->ToLog("Downloading starts. Count links: ".count($urls));
      $startTime = microtime(true);
      $cycles = 0;
      $maxThreads = 0;
      $avg = 1;
      $this->urls = array();
      $this->allStreams = array();
      do
      {
         $rStreams = $this->toRead;
         $wStreams = $this->toWrite;
         $eStreams = $this->toException;
         $num = @stream_select($rStreams, $wStreams, $eStreams, 0, 100);

         if ($num > 0)
         {
            if (count($wStreams) > 0)
            {
               foreach($wStreams as $write)
               {
                  $url =& $this->urls[array_search($write, $this->allStreams)];
                  $func = $this->supportProtocols[$url["scheme"]];
                  $this->ToLog("[{$url["num"]}]Request is sended");
                  if ($this->$func($url, MD_SEND) !== false)
                  {
                     $this->toRead[] = $write;
                  }
                  unset($this->toWrite[array_search($write, $this->toWrite)]);
               }
            }

            if (count($rStreams) > 0)
            {
               foreach($rStreams as $read)
               {
                  $url =& $this->urls[array_search($read, $this->allStreams)];
                  $this->ToLog("[{$url["num"]}]Getting data");
                  $func = $this->supportProtocols[$url["scheme"]];
                  if ($this->$func($url, MD_READ))
                  {
                     unset($this->toRead[array_search($read, $this->toRead)]);
                     unset($this->toException[array_search($read, $this->toException)]);
                  }
               }
            }

            if (count($eStreams) > 0)
            {
               foreach($eStreams as $exception)
               {
                  $url =& $this->urls[array_search($exception, $this->allStreams)];
                  $func = $this->supportProtocols[$url["scheme"]];
                  $this->$func($url, MD_EXCPECT);
                  unset($this->toRead[array_search($exception, $this->toRead)]);
                  unset($this->toWrite[array_search($exception, $this->toWrite)]);
                  unset($this->toException[array_search($exception, $this->toException)]);
               }
            }
         }
         elseif (count($this->toRead) < $this->maxThread && count($urls) > 0)
         {
            $newUrl = array_shift($urls);
            $this->ToLog("Creation of new stream[$num] for URL $newUrl");
            $num = $this->CreateStream($newUrl);
         }
         $threads = count($this->toRead);
         $cycles++;
         $maxThreads = $threads > $maxThreads ? $threads : $maxThreads;
         $avg += $threads;
         $this->stats[] = count($this->toRead);
      } while(count($this->toRead) > 0 || count($this->toWrite) > 0 || count($urls) > 0);

      $this->ToLog("Downloading is successfully completed. Elapsed ".(microtime(true) - $startTime)." sec.");
      $this->ToLog("Cycles: $cycles. Max threads: $maxThreads. Average: ".($threads / $cycles).".");
      return $this->urls;
   }

   protected function DownloadViaHTTP(&$urlInfo, $type)
   {
      $stream = $urlInfo["stream"];
      switch ($type)
      {
         case MD_SEND:
            if (fwrite($stream, sprintf(MD_REQUEST, $urlInfo["full"], $urlInfo["host"])) === false)
            {
               $this->ToLog("[{$urlInfo["num"]}]Error of sending request");
               return false;
            }
            break;
         case MD_READ:
            if ($urlInfo["code"] == 0)
            {
               stream_set_timeout($stream, $this->timeout);
               $str = trim(fgets($stream, 1024));
               $code = explode(" ", $str);
               if (isset($code[1]))
               {
                  $urlInfo["code"] = (int)$code[1];
               }
               $str = fgets($stream, 1024);
               while ($str !== false && $str != "\r\n")
               {
                  $header = explode(":", $str);
                  $urlInfo["headers"][strtolower(trim($header[0]))] = strtolower(trim($header[1]));
                  $str = fgets($stream, 1024);
               }
            }
            if (isset($urlInfo["headers"]["transfer-encoding"]) && $urlInfo["headers"]["transfer-encoding"] == "chunked")
            {
               return $this->ReadChanked($urlInfo);
            }
            else
            {
               $this->ToLog("[{$urlInfo["num"]}]Reading of all content... ", false);
               while (!feof($stream)) $urlInfo["contents"] .= fread($stream, 8192);
               $this->ToLog("Ok");
            }
            break;
         case MD_EXCPECT:
            $this->ToLog("[{$urlInfo["num"]}]An unknown error occurred");
            break;
      }
      return true;
   }

   public function ReadChanked(&$urlInfo)
   {
      $stream = $urlInfo["stream"];
      do
      {
         $len = fgets($stream);
         $meta = stream_get_meta_data($stream);
         if ($len === false || $meta["unread_bytes"] == 0/* || trim($len) == ''*/)
         {
            return false;
         }
         if (trim($len) == '')
         {
            $len = fgets($stream);
         }

         if (!preg_match("|^[0-9a-f]+\s*$|i", $len))
         {
            $this->ToLog("[{$urlInfo["num"]}]Error: Unknown block length");
            return true;
         }
         else
         {
            $len = hexdec(trim($len));
            if ($len == 0)
            {
               break;
            }
         }
         if ($len > 0)
         {
            stream_set_blocking($stream, 1);
            $this->ToLog("[{$urlInfo["num"]}]Reading of $len byte... ", false);
            while($len > 0)
            {
               $buffer = fread($stream, $len < 1024 ? $len : 1024);
               if ($buffer === false)
               {
                  $this->ToLog("Fail");
                  return true;
               }
               $len -= strlen($buffer);
               $urlInfo["contents"] .= $buffer;
            }
            $this->ToLog("Ok");
            stream_set_blocking($stream, 0);
         }
      } while(true);
      return true;
   }

   public function __destruct()
   {
      foreach ($this->allStreams as $stream)
      {
         fclose($stream);
      }
      $this->allStreams = null;

      foreach ($this->logs as $stream)
      {
         @fclose($stream);
      }
      $this->logs = null;
   }
}

?>